package org.example.cluster;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.corundumstudio.socketio.SocketIOClient;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;
import org.example.cache.ClientCache;
import org.example.po.Msg;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplicationShutdownHandlers;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import java.io.UnsupportedEncodingException;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;

/**
 * @author zhangjg
 * @desc c
 * @date 2023/05/07 18:33
 **/
@Component
@Slf4j
public class SubClient implements InitializingBean {

    @Autowired
    private ClientCache clientCache;

    @Autowired
    private DefaultMQPushConsumer consumer;

    @Override
    public void afterPropertiesSet() throws Exception {
        consumer.start();
    }

    @Bean
    public DefaultMQPushConsumer getConsumer() {
        // 初始化consumer，并设置consumer group name
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("cg_current");
        // 设置NameServer地址
        consumer.setNamesrvAddr("192.168.10.1:9876");
        consumer.setMessageModel(MessageModel.BROADCASTING);
        //订阅一个或多个topic，并指定tag过滤条件，这里指定*表示接收所有tag的消息
        try {
            consumer.subscribe("t11", "*");
        } catch (Exception e) {
            log.error(e.getMessage(),e);
        }
        //注册回调接口来处理从Broker中收到的消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                // 返回消息消费状态，ConsumeConcurrentlyStatus.CONSUME_SUCCESS为消费成功
                return boardCastMsg(msgs);
            }
        });

        return consumer;
    }


    private ConsumeConcurrentlyStatus boardCastMsg(List<MessageExt> msgs){
        Map<String, HashMap<UUID, SocketIOClient>> all = clientCache.getAll();
        for (MessageExt msg : msgs){
            Msg message = JSON.parseObject(new String(msg.getBody(), StandardCharsets.UTF_8), Msg.class);
            log.info("收到消息，sid={},userid={},msg={},dest={}", message.getSessionId(),message.getUserId(), message.getMessage(),message.getDest());
            if(StringUtils.isEmpty(message.getDest())){
//                if (all.get(message.getDest()) != null){
                    all.forEach((key,hash) -> {
                        hash.forEach((uuid,client) -> {
                            client.sendEvent("chatevent", message);
                        });
                    });
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
//                }else {
//                }
            }else {
                HashMap<UUID, SocketIOClient> map = all.get(message.getDest());
                if(map != null){
                    map.forEach((uuid,client) -> {
                        client.sendEvent("chatevent", message);
                    });
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }else {
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            }
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }

}
